package com.demo.java.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.Iterator;

import static com.demo.java.utils.LogUtils.log;

public class SingleThreadReactorServer {
    public static void main(String[] args) throws IOException, InterruptedException {
        Thread thread = new Thread(new Reactor());
        thread.start();
        thread.join();
    }

    static class Reactor implements Runnable {
        private Selector selector;
        private ServerSocketChannel serverSocketChannel;

        public Reactor() throws IOException {
            this.selector = Selector.open();
            this.serverSocketChannel = ServerSocketChannel.open();
            this.serverSocketChannel.configureBlocking(false);

            ServerSocket socketChannel = serverSocketChannel.socket();
            socketChannel.bind(new InetSocketAddress(7891));
            System.out.println("server start listen on: " + 7891);
//            this.serverSocketChannel.bind(new InetSocketAddress(7891));

            final SelectionKey selectionKey = serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
            selectionKey.attach(new AcceptorHandler(selectionKey, this.selector));
        }

        @Override
        public void run() {
            try {
                while (this.selector.select() > 0) {
                    final Iterator<SelectionKey> keyIterator = this.selector.selectedKeys().iterator();
                    while (keyIterator.hasNext()) {
                        final SelectionKey sk = keyIterator.next();
                        keyIterator.remove();

                        if (sk.isAcceptable()) {
                            ((AcceptorHandler) sk.attachment()).run();
                        } else {
                            ((EchoHandler) sk.attachment()).setSk(sk).run();
                        }
                    }
                }
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    static class AcceptorHandler implements Runnable {
        private SelectionKey selectionKey;
        private Selector selector;
        public AcceptorHandler(SelectionKey selectionKey, Selector selector) {
            this.selectionKey = selectionKey;
            this.selector = selector;
        }
        public void run() {
            System.out.println(log("accept"));
            try {
                final SocketChannel socketChannel = ((ServerSocketChannel) selectionKey.channel()).accept();
                socketChannel.configureBlocking(false);
                socketChannel.register(selector, SelectionKey.OP_READ).attach(new EchoHandler(selector));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    static class EchoHandler implements Runnable {
        private SelectionKey sk;
        private Selector selector;

        public EchoHandler(Selector selector) {
            this.selector = selector;
        }
        public void run() {
//            System.out.println("echo handler");
            final ByteBuffer buffer = ByteBuffer.allocate(1024);
            SocketChannel channel = (SocketChannel) sk.channel();
            if (this.sk.isReadable()) {
//                System.out.println(log("run read"));
                try {
                    int length = 0;
                    StringBuilder sb = new StringBuilder();
                    Charset charset = Charset.forName("utf-8");
                    while ((length = channel.read(buffer)) > 0) {
                        // 切换读写模式，这里是从channel往buffer中写，所以切换后为读模式, position=0, limit=position写
                        buffer.flip();
                        // 将Buffer转换为String的方法
                        sb.append(charset.decode(buffer).toString());
                        buffer.clear();
                    }
                    // 清空ByteBuffer， position=0，limit=capacity
                    buffer.clear();
                    System.out.println(log("read: " + sb.toString()));
                    if (length > 0) {
                        // 将当前selectionKey设置为关注写事件
                        this.sk.interestOps(SelectionKey.OP_WRITE);
                    }
                    // 当客户端关闭时，服务端会读到数据长度为-1，表示客户端关闭
                    if (length == -1) {
                        System.out.println("client closed");
                        channel.close();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            } else if (this.sk.isWritable()) {
                System.out.println(log("run write"));

                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
//            System.out.println(log("echo run read/write"));
        }

        public EchoHandler setSk(SelectionKey sk) {
            this.sk = sk;
            return this;
        }
    }
}
